package com.zhang.third.day03;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

/**
 * @title: rebalance和rescale对比
 * @author: zhang
 * @date: 2022/4/6 11:19
 */
public class Example6 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env
                .addSource(new RichParallelSourceFunction<String>() {
                    @Override
                    public void run(SourceContext<String> ctx) throws Exception {
                        for (int i = 1; i < 9; i++) {
                            if (i%2==getRuntimeContext().getIndexOfThisSubtask()){
                                ctx.collect("并行子任务：" + getRuntimeContext().getIndexOfThisSubtask() + "" +
                                        ",发送的数据是：" + i);
                            }
                        }
                    }

                    @Override
                    public void cancel() {

                    }
                })
                .setParallelism(2)
                .rebalance()
                .print().setParallelism(4);


        env
                .addSource(new RichParallelSourceFunction<String>() {
                    @Override
                    public void run(SourceContext<String> ctx) throws Exception {
                        for (int i = 1; i < 9; i++) {
                            if (i%2==getRuntimeContext().getIndexOfThisSubtask()){
                                ctx.collect("并行子任务：" + getRuntimeContext().getIndexOfThisSubtask() + "" +
                                        ",发送的数据是：" + i);
                            }
                        }
                    }

                    @Override
                    public void cancel() {

                    }
                })
                .setParallelism(2)
                .rescale()
                .print().setParallelism(4);

        env.execute();
    }
}
